package com.navinfo.tripanalysis.offline.service;

import com.navinfo.tripanalysis.common.pojo.Point;
import com.navinfo.tripanalysis.common.pojo.PointEngine;
import com.navinfo.tripanalysis.common.pojo.ProtocolEnum;
import com.navinfo.tripanalysis.common.util.CommonUtils;
import com.navinfo.tripanalysis.offline.pojo.LoadDataParam;
import org.apache.commons.collections.CollectionUtils;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;

import java.util.*;

/**
 * 加载发动机数据的基本服务
 * @author 刘俊
 */
public abstract class AbstractLoadEngineDataServiceBigData implements LoadEngineDataService {
    /**
     * 协议中是否含有protocoltype字段
     */
    protected  boolean supportProtocolType;
    private final String PROTOCOL_NAME = "protocoltype";

    protected static final List<String> COL_NAMES = new ArrayList<>();

    public AbstractLoadEngineDataServiceBigData() {
        COL_NAMES.addAll(Arrays.asList("gpstime","vehiclespeed","barometricpressure","engineoutputtorque","frictiontorque",
                "rotation","fuelconsumptionrate","scrupstreamnoxsensoroutput","scrdownstreamnoxsensoroutput","reagentallowance",
                "airinflow","scrinlettemperature","scroutlettemperature","dpfdropoutvoltage","enginecoolanttemperature",
                "tanklevel","positioningstate","longitude","latitude","mileage","vin",
                "ternarycatalystdownstreamnoxsensoroutput","ternarycatalystupstreamo2sensoroutput","ternarycatalystdownstreamo2sensoroutput","ternarycatalyticconvertertemperature"));
    }

    public boolean isSupportProtocolType() {
        return supportProtocolType;
    }

    public void setSupportProtocolType(boolean supportProtocolType) {
        if (supportProtocolType) {
            if (!COL_NAMES.contains(PROTOCOL_NAME)) {
                COL_NAMES.add(PROTOCOL_NAME);
            }
        } else {
            if (COL_NAMES.contains(PROTOCOL_NAME)) {
                COL_NAMES.remove(PROTOCOL_NAME);
            }
        }
        this.supportProtocolType = supportProtocolType;
    }

    /**
     * 获取过滤条件
     * @param param
     * @return
     */
    protected String getWhereSqlEngine(LoadDataParam param) {
        String whereSql = "gpstime != 0 and gpstime is not null and tid != 0 and tid is not null ";

        List<Long> tids = param.getTidList();
        if (CollectionUtils.isNotEmpty(tids)) {
            StringBuilder sb = new StringBuilder();
            sb.append(String.format(" and tid in(%s) ", CommonUtils.mkString(tids, ",")));
            whereSql += sb.toString();
        }

        //过滤出原发动机数据流协议类型，这块可以去掉，因为表已经单独拆分出来了
        if(isSupportProtocolType()){
            whereSql += " and protocoltype=3 ";
        }

        return whereSql;
    }

    protected JavaRDD<Point> toJavaRdd(Dataset<Row> dataset) {
        return dataset.toJavaRDD().mapPartitions((FlatMapFunction<Iterator<Row>, Point>) iterator ->
                new Iterator<Point>() {
                    @Override
                    public boolean hasNext() {
                        return iterator.hasNext();
                    }
                    @Override
                    public Point next() {
                        return getPoint(iterator.next());
                    }
                }
        );
    }

    private Point getPoint(Row row) {
        PointEngine point = new PointEngine();
        point.setTid((long) Optional.ofNullable(row.getAs("tid")).orElse(0));
        point.setProtocol(ProtocolEnum.ENGINE);
        point.setGpsTime((long) Optional.ofNullable(row.getAs("gpstime")).orElse(0));
        point.setVehicleSpeed((int) Optional.ofNullable(row.getAs("vehiclespeed")).orElse(0) / 100);
        point.setBarometricPressure((int) Optional.ofNullable(row.getAs("barometricpressure")).orElse(0) / 100);
        point.setEngineOutputTorque((int) Optional.ofNullable(row.getAs("engineoutputtorque")).orElse(0) / 100);
        point.setFrictionTorque((int) Optional.ofNullable(row.getAs("frictiontorque")).orElse(0) / 100);
        point.setRotation((int) Optional.ofNullable(row.getAs("rotation")).orElse(0) / 100);
        point.setFuelConsumption((int) Optional.ofNullable(row.getAs("fuelconsumptionrate")).orElse(0) / 100);
        point.setScrUpstreamNOxSensorOutput((int) Optional.ofNullable(row.getAs("scrupstreamnoxsensoroutput")).orElse(0) / 100);
        point.setScrDownstreamNOxSensorOutput((int) Optional.ofNullable(row.getAs("scrdownstreamnoxsensoroutput")).orElse(0) / 100);
        point.setReagentAllowance((int) Optional.ofNullable(row.getAs("reagentallowance")).orElse(0) / 100);
        point.setAirInflow((int) Optional.ofNullable(row.getAs("airinflow")).orElse(0) / 100);
        point.setScrInletTemperature((int) Optional.ofNullable(row.getAs("scrinlettemperature")).orElse(0) / 100);
        point.setScrOutletTemperature((int) Optional.ofNullable(row.getAs("scroutlettemperature")).orElse(0) / 100);
        point.setDpfDropoutVoltage((int) Optional.ofNullable(row.getAs("dpfdropoutvoltage")).orElse(0) / 100);
        point.setEngineCoolantTemperature((int) Optional.ofNullable(row.getAs("enginecoolanttemperature")).orElse(  0) / 100);
        point.setTankLevel((int) Optional.ofNullable(row.getAs("tanklevel")).orElse(0) / 100);
        //定位状态：为1时，是无效定位
        point.setPositioningState((int) Optional.ofNullable(row.getAs("positioningstate")).orElse(1) / 100);
        point.setLongitude((int) Optional.ofNullable(row.getAs("longitude")).orElse(0));
        point.setLatitude((int) Optional.ofNullable(row.getAs("latitude")).orElse(0));
        point.setMileage(Long.parseLong(String.valueOf(Optional.ofNullable(row.getAs("mileage")).orElse(0))) / 100);
        point.setVin((String) Optional.ofNullable(row.getAs("vin")).orElse(0));
        point.setTernaryCatalystDownstreamNOxSensorOutput((int) Optional.ofNullable(row.getAs("ternarycatalystdownstreamnoxsensoroutput")).orElse(0)/ 100);
        point.setTernaryCatalystUpstreamO2SensorOutput((int) Optional.ofNullable(row.getAs("ternarycatalystupstreamo2sensoroutput")).orElse(0)/ 100);
        point.setTernaryCatalystDownstreamO2SensorOutput((int) Optional.ofNullable(row.getAs("ternarycatalystdownstreamo2sensoroutput")).orElse(0)/ 100);
        point.setTernaryCatalyticConverterTemperature((int) Optional.ofNullable(row.getAs("ternarycatalyticconvertertemperature")).orElse(0)/ 100);
        return point;
    }
}